#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：henan_sumo
@File ：main_send.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/4/16 17:48
'''
import threading
from copy import copy
from multiprocessing import Queue, Process
from loguru import logger
import time
import json
from kafka import KafkaConsumer, KafkaProducer

from config.config import KAFKA_HOST
from utils.config_temp import RADAR_DATA_LIST_CJ, RADAR_DATA_LIST_JJ, first_CJ, first_JJ
from utils.tools import time_str


class DataFusion:
    def __init__(self, receive_topic, org_code_list, send_radar_topic, send_kako_topic, RADAR_DATA_LIST, first_sn):
        self.RADAR_DATA_LIST = RADAR_DATA_LIST
        self.first_sn_list = first_sn
        self.receive_topic = receive_topic
        self.org_code_list = org_code_list
        self.send_radar_topic = send_radar_topic
        self.send_kako_topic = send_kako_topic
        self.topic_dict = {}
        self.plate_sn_dict = {}
        self.kako_cache_dict = {}
        self.init_kafka()

    def init_kafka(self):
        for sn in self.RADAR_DATA_LIST.keys():
            self.topic_dict[sn] = None
            self.kako_cache_dict[sn] = {}
            self.plate_sn_dict[sn] = {}

        threading.Thread(target=self.receive_data, args=(receive_topic,)).start()

    # def start_send(self):
    #     threading.Thread(target=self.send_radar_kafka).start()
    #     threading.Thread(target=self.send_kako_kafka).start()

    def receive_data(self, kafka_topic):
        kafka_consumer_e1 = KafkaConsumer(
            kafka_topic,
            bootstrap_servers=KAFKA_HOST,
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                                       value_serializer=lambda x: json.dumps(x).encode('utf-8'))
        while True:
            time.sleep(0.1)
            target_data = []
            target_result = {}
            for message in kafka_consumer_e1:
                if message:
                    try:

                        message_value = message.value.decode('utf-8')
                        kako_info = json.loads(message_value)
                        station = kako_info['stationID']
                        car_id = kako_info['vehicleId']
                        target_info = {'id': car_id, 'station': station,
                                       'lon': RADAR_DATA_LIST_CJ[station]['lon'],
                                       'lat': RADAR_DATA_LIST_CJ[station]['lat'], 'angle': 0,
                                       'carType': 1, 'picLicense': kako_info['picLicense'],
                                       'lane': kako_info['laneNum'], 'speed': 80,
                                       'orgCode': '', 'timestamp': int(time.time() * 1000),
                                       'global_id': str(kako_info['vehicleId']) + "-" + str(station)}
                        # {'id': 600011, 'station': 17, 'lon': 116.428959, 'lat': 39.745319, 'angle': 168.4, 'carType': 1, 'picLicense': '鄂IV8396', 'lane': 3, 'speed': 103, 'orgCode': 'CJ-10-02','timestamp':1750145338600,'global_id':'600011_17'}
                        if station in self.topic_dict.keys():
                            if station in self.first_sn_list:
                                picLicense = kako_info['picLicense']
                                if "默A" not in picLicense:
                                    self.plate_sn_dict[station][picLicense] = [int(time.time()),
                                                                               target_info['global_id']]
                                self.kako_cache_dict[station][car_id] = int(time.time())
                                logger.info(f"卡口上报{target_info}")
                                kafka_producer.send(self.send_kako_topic, value=target_info,
                                                    key=str(int(time.time() * 1000)))
                            else:

                                picLicense = target_info['picLicense']
                                for first_sn in self.first_sn_list:
                                    if picLicense in self.plate_sn_dict[first_sn].keys():
                                        target_info['timestamp'] = time_str(int(time.time() * 1000))
                                        target_info['global_id'] = self.plate_sn_dict[first_sn][picLicense][1]
                                        logger.info(f"卡口上报{target_info}")
                                        kafka_producer.send(self.send_kako_topic, value=target_info,
                                                            key=str(int(time.time() * 1000)))
                    except Exception as e:
                        logger.info(f"kafka数据处理异常{e}")


if __name__ == '__main__':
    topic_list = ['kako_data_CJ', 'kako_data_JJ']
    org_code_list = ['kako_data_CJ', 'kako_data_JJ']
    send_radar_topic = "fusion_E1_data"
    send_kako_topic = "fusion_kako_data"
    receive_topic = "CJ-10-SUM"
    data_fusion = DataFusion(receive_topic, org_code_list, send_radar_topic, send_kako_topic, RADAR_DATA_LIST_CJ,
                             first_CJ)
    data_fusion.start_send()
